package com.atguigu.edu.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TableProcess;
import com.atguigu.edu.common.EduConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class MyBoradCastFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {

    //定义状态描述器
    private MapStateDescriptor<String, TableProcess> tableConfigDescriptor;
    // 定义Phoenix的连接
    private Connection conn;

    //运用构造器定义构造器
    public MyBoradCastFunction(MapStateDescriptor<String,TableProcess> tableConfigDescriptor){
        this.tableConfigDescriptor=tableConfigDescriptor;
    }

    /**
     * 在open方法中获取数据的链接
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        //加载Phoenix数据库驱动
        Class.forName(EduConfig.PHOENIX_DRIVER);
        //获取链接
        DriverManager.getConnection(EduConfig.PHOENIX_SERVER);
    }

    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //获取广播流的状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(tableConfigDescriptor);
        //获取数据中表名
        String table = jsonObject.getString("table");
        //根据表名获取广播状态中的数据
        TableProcess tableProcess = broadcastState.get(table);
        //根据获取结果判断该表是否为维度表
        if(tableProcess!=null){
            //获取该表的数据
            JSONObject data = jsonObject.getJSONObject("data");
            //获取操作类型
            String type = jsonObject.getString("type");
            //根据sinkColumns过滤字段数据
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumns(data,sinkColumns);
            //获取目标表名
            String sinkTable = tableProcess.getSinkTable();
            //将目标表名加入到主流中
            data.put("sinkTable",sinkTable);
            //将操作类型加入到主流中
            data.put("type",type);
            collector.collect(data);

        }

    }

    private void filterColumns(JSONObject data, String sinkColumns) {
        //将sinKColumns转换成数组
        String[] sinkColumnsArr = sinkColumns.split(",");
        //将sinkColumnsArr转换成列表
        List<String> columnsList = Arrays.asList(sinkColumnsArr);
        //遍历data字段判断是否包含在columnsList
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            if(!columnsList.contains(next.getKey())){
                //不包含则移除该字段
                iterator.remove();
            }
        }
    }

    @Override
    public void processBroadcastElement(String s, Context context, Collector<JSONObject> collector) throws Exception {
        //将广播流的数据结构转换成JsonObject
        JSONObject jsonObject = JSON.parseObject(s);
        //获取广播流的状态
        BroadcastState<String, TableProcess> tableProcessBroadcastState = context.getBroadcastState(tableConfigDescriptor);
        //获取官博流数据的操作类型
        String op = jsonObject.getString("op");
        //如果是删除操作
        if("d".equals(op)){
            //获取删除的数据
            TableProcess before = jsonObject.getObject("before", TableProcess.class);
            //获取维度表名
            String sourceTable = before.getSourceTable();
            //将状态中的维度表删除
            tableProcessBroadcastState.remove(sourceTable);
        }else {//其他操做类型，则需要获取数据信息并更新状态和在Phoenix建立表
            //获取配置表数据
            TableProcess after = jsonObject.getObject("after", TableProcess.class);
            //获取配置表的字段
            String sourceTable = after.getSourceTable();
            String sinkTable = after.getSinkTable();
            String sinkColumns = after.getSinkColumns();
            String sinkPk = after.getSinkPk();
            String sinkExtend = after.getSinkExtend();
            //将维度表信息保存在状态中
            tableProcessBroadcastState.put(sourceTable,after);
            //在Phoenix建立维度表
            checkTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
        }
    }

    /**
     * 在Phoenix建立维度表
     * @param sinkTable
     * @param sinkColumns
     * @param sinkPk
     * @param sinkExtend
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        //封装建立维度表的sql
        StringBuilder sql = new StringBuilder();
        sql.append("create table if not exists " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "(\n");
        //将sinkColumns转换成数组
        String[] sinkColumnsArr = sinkColumns.split(",");
        //为主键设置默认值
        if(sinkPk==null){
            sinkPk="id";
        }
        //为额外字段设置默认值
        if(sinkExtend==null){
            sinkExtend="";
        }
        //遍历sinkColumnsArr获取建表字段并完善建表sql语句
        for(int i=0;i<sinkColumnsArr.length;i++){
            sql.append(sinkColumnsArr[i]+" varchar");
            //判断当前字段是否为主键
            if(sinkColumnsArr[i].equals(sinkPk)){
                sql.append(" primary key");
            }
            //判断是否为最后一个字段,不是则执行
            if(i<sinkColumnsArr.length-1){
                sql.append(",\n");
            }
        }
        //补充完整最后的建表信息
        sql.append(")");
        sql.append(sinkExtend);
        //将建表语句转换成String类型
        String createStatement = sql.toString();
        //为数据操作对象赋默认值
        PreparedStatement preparedStatement=null;
        try {
            //获取数据库操作对象
            preparedStatement = conn.prepareStatement(createStatement);
            //执行sql
            preparedStatement.execute();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            System.out.println("建表语句\n" + createStatement + "\n执行异常");
        } finally {
            if(preparedStatement!=null){
                try {
                    //关闭链接
                    preparedStatement.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                    throw new RuntimeException("数据库操作对象释放异常");
                }
            }
        }
    }
}
